package com.example;

import com.example.constant.KafkaTopic;
import com.example.entity.User;
import com.google.common.primitives.Bytes;
import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;
import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.state.KeyValueStore;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * KafkaStreamUser
 *
 * @author yongxiang
 * @date 2024/9/3 14:50
 * @desc
 */
public class KafkaStreamUser {
    public static void main(String[] args) {

        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "kafka-streams-example");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.8.80:9092,192.168.8.80:9093,192.168.8.80:9094");
        props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://192.168.8.80:8081");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class.getName());

        StreamsBuilder builder = new StreamsBuilder();
        Map<String, String> serdeConfig = new HashMap<>();
        serdeConfig.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://192.168.8.80:8081");

        final SpecificAvroSerde<User> userSerde = new SpecificAvroSerde<>();
        userSerde.configure(serdeConfig, false);


        KStream<String, User> stream = builder.stream(KafkaTopic.userTopic, Consumed.with(Serdes.String(),userSerde));
        stream.toTable(Materialized.<String, User, KeyValueStore<org.apache.kafka.common.utils.Bytes, byte[]>>as(KafkaTopic.userStore)
                .withKeySerde(Serdes.String())
                .withValueSerde(userSerde));


        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
    }
}
